package com.ly.integral.listener;

import com.alibaba.fastjson.JSONObject;
import com.ly.integral.service.IntegralService;
import com.domain.Order;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;


//消费者监听器
//监听到消息之后，调用业务服务类处理即可。
// 处理完成则返回CONSUME_SUCCESS以提交，处理失败则返回RECONSUME_LATER来重试
@Component
public class IntegralListener implements MessageListenerConcurrently {

    @Autowired
    IntegralService integralService;
//    Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
//        logger.info("消费者线程监听到消息。");
        System.out.println("消费者线程监听到消息。");
        try{
            for (MessageExt message:list) {
//                logger.info("开始处理订单数据，准备增加积分....");
                Order order  = JSONObject.parseObject(message.getBody(), Order.class);
                integralService.insert(order);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
//            logger.error("处理消费者数据发生异常。{}",e);
            System.out.println("处理消费者数据发生异常。{}"+e.getMessage());
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }


}
